package com.wzf.greattruth.test.zookeeper.queue;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.SortedSet;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;

import com.alibaba.fastjson.JSON;
import com.wzf.greattruth.zookeeper.ZookeeperManager;
import com.wzf.greattruth.zookeeper.node.Node;
import com.wzf.greattruth.zookeeper.queue.DistributedQueue;

public class ZookeeperQueueTester {
    private static String HOST = "localhost:2181";
    private static int SESSION_TIMEOUT = 1000 * 60 * 60;

    protected static final List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;

    private static ZooKeeper zookeeper;

    public static void main(String[] args) throws Exception {
        zookeeper = new ZooKeeper(HOST, SESSION_TIMEOUT, null);

        ZookeeperQueueTester tester = new ZookeeperQueueTester();
         tester.testQueue();
//        tester.test();
        // zookeeper.close();

    }

    private void test() {
//        String path = File.separator + "zk" + File.separator + "a" + File.separator + "b" + File.separator;
//        ZookeeperManager.getInstance().createDirIfNotExist(zookeeper, path);
        String path = File.separator + "zk" + File.separator + "a2/" ;
        ZookeeperManager.getInstance().createDir(zookeeper, path, null, CreateMode.PERSISTENT) ;


    }

    private void testQueue() {
        int times = 10;
        final String queueDir = File.separator + "zk" + File.separator + "queue";
        for (int i = 0; i < times; i++) {
            final int index = i;
            Thread t = new Thread(new Runnable() {

                @Override
                public void run() {
                    DistributedQueue queue = new DistributedQueue(zookeeper, queueDir);
                    String value = index + "-" + Thread.currentThread().getId();
                    String rs = queue.offer(value.getBytes());
                    System.err.println(rs + "  =  " + value);
                }
            });
            t.start();
        }
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.err.println("===============children===============");
        String prefix = "que-";
        SortedSet<Node> nodes = ZookeeperManager.getInstance().getChildren(zookeeper, queueDir, prefix);
        for (Node node : nodes) {
            System.err.println(node.getId());
        }

        System.err.println("===============take from queue===============");
        DistributedQueue queue = new DistributedQueue(zookeeper, queueDir);
        while (true) {
            byte[] datas = queue.take();
            if (datas == null) {
                break;
            } else {
                String value = new String(datas);
                System.err.println(value);
            }
        }
    }

}
